package com.dayu.finecomm.device;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.timeout.ReadTimeoutException;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dayu.finecomm.pojo.DeviceConfig;
import com.dayu.finecomm.protocol.codec.Drive;
import com.dayu.finecomm.source.CollectSource;
import com.dayu.finecomm.utils.Helpers;

/**
 * 水利局设备
 * 
 * @author Lijiajie
 * 
 */
@Drive(SljDevice.DRIVE)
public class SljDevice extends TcpDevice {

	static final String DRIVE = "slj";

	static Logger LOGGER = LoggerFactory.getLogger(SljDevice.class);

	Map<String, byte[]> IDs;

	public SljDevice(CollectSource stations, Set<String> protocols, DeviceConfig config) {

		super(stations, protocols, config);

		if (this.config.idle == 0) {
			this.config.idle = 600;
		}

		IDs = new ConcurrentHashMap<>();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {

		IDs.clear();
		super.channelInactive(ctx);
	}

	private String getSessionId(ChannelHandlerContext ctx) {
		return Helpers.host(ctx);
	}

	@Override
	protected void initChannel(NioSocketChannel ctx) throws Exception {

		super.initChannel(ctx);

		ChannelPipeline pipeline = ctx.pipeline();

		pipeline.addFirst(new ByteToMessageDecoder() {

			@Override
			public void channelInactive(ChannelHandlerContext ctx) throws Exception {

				IDs.remove(getSessionId(ctx));
				super.channelInactive(ctx);
			}

			@Override
			public void channelActive(ChannelHandlerContext ctx) throws Exception {

				ctx.channel().config().setOption(ChannelOption.SO_BACKLOG, 128);
				ctx.channel().config().setOption(ChannelOption.TCP_NODELAY, true);
				ctx.channel().config().setOption(ChannelOption.SO_KEEPALIVE, true);

				super.channelActive(ctx);
			}

			@Override
			protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

				boolean invalid = false;

				if (in.getUnsignedByte(0) != 0x24) {
					invalid = true;
				}

				int len = in.getUnsignedByte(1) + in.getUnsignedByte(2) * 256;

				if (len < in.readableBytes()) {
					invalid = true;
				}

				if (invalid) {

					byte[] ID = IDs.get(getSessionId(ctx));

					if (ID != null) {
						LOGGER.info("Close invalid remote pair: " + " " + new String(ID));
					}

					IDs.remove(getSessionId(ctx));
					ctx.channel().close();

				} else {

					register(ctx, in);
					unregister(ctx, in);
					heartbeat(ctx, in);
					data(ctx, in);

					in.skipBytes(in.readableBytes() < len ? in.readableBytes() : len);
				}

			}

			private boolean data(ChannelHandlerContext ctx, ByteBuf in) {

				if (in.getUnsignedByte(3) == 0x47) {

					byte[] ID = IDs.get(getSessionId(ctx));

					if (ID != null) {

						LOGGER.info("new data packet from ZJSLJ device " + " " + new String(ID));

						int length = in.getUnsignedByte(2) * 256 + in.getUnsignedByte(1);

						if (length < 1024) {

							byte[] b = new byte[length];
							in.getBytes(0, b);
						}

						ctx.writeAndFlush(Unpooled.copiedBuffer(new byte[] { 0x24, 0x05, 0x00, 0x48, 0x73 }));

						int size = in.readableBytes() + 16;
						ByteBuf nb = ctx.alloc().buffer(size, size);

						nb.writeBytes(new byte[] { 'I', 'D', ':' });
						nb.writeBytes(ID);
						nb.writeBytes(new byte[] { ';' });
						nb.writeBytes(in);
						nb.readerIndex(0);

						ctx.fireChannelRead(nb);

						return true;
					}
				}

				return false;
			}

			private boolean heartbeat(ChannelHandlerContext ctx, ByteBuf in) {

				// 心跳帧
				if (in.getUnsignedByte(3) != 0x45)
					return false;

				byte[] ID = IDs.get(getSessionId(ctx));

				if (ID != null) {
					LOGGER.info("HB device " + " " + new String(ID) + " " + Helpers.host(ctx));
				}

				ctx.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 0x24, 0x05, 0x00, 0x46, 0x73 }));

				return true;
			}

			private boolean unregister(ChannelHandlerContext ctx, ByteBuf in) {

				// 注销帧
				if (in.getUnsignedByte(3) != 0x43)
					return false;

				IDs.remove(getSessionId(ctx));

				ctx.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 0x24, 0x05, 0x00, 0x44, 0x73 }));

				return true;
			}

			private boolean register(ChannelHandlerContext ctx, ByteBuf in) {

				// 注册帧
				if (in.getUnsignedByte(3) != 0x41)
					return false;

				byte[] ID = new byte[11];
				in.getBytes(4, ID);

				LOGGER.info("new ZJSLJ device " + " " + new String(ID) + " " + Helpers.host(ctx));

				String sessionId = getSessionId(ctx);
				IDs.put(sessionId, ID);

				ctx.writeAndFlush(Unpooled.wrappedBuffer(new byte[] { 0x24, 0x05, 0x00, 0x42, 0x73 }));

				return true;
			}

		});

		// Finalize
		pipeline.addLast(new ChannelDuplexHandler() {

			@Override
			public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

				if (cause instanceof ReadTimeoutException) {
					IDs.remove(getSessionId(ctx));
				} else if (cause instanceof IOException) {
					LOGGER.info("Force close by remote: " + " " + new String(IDs.get(getSessionId(ctx))));
					IDs.remove(getSessionId(ctx));
				} else {
					super.exceptionCaught(ctx, cause);
				}
			}
		});
	}
}